package com.yonyou.common.service.messagecenter.rabbitmq;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Address;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.yonyou.common.service.messagecenter.config.RabbitMQConfig;
import com.yonyou.common.service.messagecenter.datamapping.DataMapping;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.ConsumerEntity;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.ExchangeEntity;
import com.yonyou.common.service.messagecenter.rabbitmq.entity.QueueEntity;
import com.yonyou.common.service.messagecenter.util.MqConsts;

import net.sf.json.JSONArray;

@Configuration
@Component
public class MQSettingService implements InitializingBean,DisposableBean, ApplicationListener<ContextRefreshedEvent> {

	private static Logger logger=Logger.getLogger(MQSettingService.class);
    @Autowired
    private RabbitMQConfig config;
    @Autowired
    private ApplicationContext context;
    private List<ExchangeEntity> listExchange;
    private List<QueueEntity> listQueue;
    private List<ConsumerEntity> listConsumer;
    private ConnectionFactory factory;
    private Connection connection;
    private Connection ProduceConnection;
    private boolean inited=false;
	@Bean
    public ConnectionFactory rabbitMQFactory(){
    	return getConnectionFactory();
    }
    @Bean("consumerConnection")
    public Connection rabbitMQConnection() throws IOException, TimeoutException{
    	return getConnection();
    }
    @Bean("produceConnection")
    public Connection rabbitMQConnectionProduce() throws IOException, TimeoutException{
    	return getProduceConnection();
    }

    @Bean("connectionFactory")
	public ConnectionFactory getConnectionFactory(){
		if(factory==null)
           factory = new ConnectionFactory();
		return factory;
	}
    
    public Connection getProduceConnection() throws IOException, TimeoutException {
		if(this.ProduceConnection==null)
			ProduceConnection = newConnection();
		return ProduceConnection;
	}
	public void setProduceConnection(Connection produceConnection) {
		ProduceConnection = produceConnection;
	}
	
	public void init(Channel channel) throws IOException{
		JSONArray array1=JSONArray.fromObject(config.getExchanges());
        listExchange=(List<ExchangeEntity>)array1.toCollection(array1,ExchangeEntity.class);
        for(ExchangeEntity e : listExchange){
        	channel.exchangeDeclare(e.getName(), e.getType(),true,false,null);
        }
        
        //声明重发direct类型的exchange
    	channel.exchangeDeclare(MqConsts.DEFAULT_DIRECT_EXCHANGE, "direct",true,false,null);
    	
		JSONArray array2=JSONArray.fromObject(config.getQueues());
		listQueue=(List<QueueEntity>)array2.toCollection(array2,QueueEntity.class);
        for(QueueEntity e : listQueue){
        	channel.queueDeclare(e.getName(), true,false,false,null);
        	channel.queueBind(e.getName(), e.getExchange(), e.getRouteKey(),null);
        	//声明默认的direct类型绑定，用于消息多次处理失败重发到本队列
        	channel.queueBind(e.getName(), MqConsts.DEFAULT_DIRECT_EXCHANGE, e.getName(),null);
        }
	}
	public boolean containQueue(String queue){
        for(QueueEntity e : listQueue){
        	if( queue.equals(e.getName()))
        		return true;
        }
		return false;
	}

	@Override
	public void destroy() throws Exception {
	}

	@Override
	public void afterPropertiesSet() throws Exception {

        factory = getConnectionFactory();
//        factory.setHost(config.getHost());
//        factory.setPort(config.getPort());
        // 关键所在，指定线程池  
//        ExecutorService service = Executors.newFixedThreadPool(100);  
//        factory.setSharedExecutor(service);  
        
        factory.setUsername(config.getUsername());
        factory.setPassword(config.getPassword());
        factory.setVirtualHost(config.getVirtualhost());
        factory.setAutomaticRecoveryEnabled(true);
        factory.setNetworkRecoveryInterval(10);
        factory.setTopologyRecoveryEnabled(false);// 设置不重新声明交换器，队列等信息。 
        connection = getConnection();
        Channel channel = connection.createChannel();
        init(channel);
	}
	private void startListener() {
        
		JSONArray array3=JSONArray.fromObject(config.getConsumers());
		listConsumer=(List<ConsumerEntity>)array3.toCollection(array3,ConsumerEntity.class);
        for(ConsumerEntity e : listConsumer){
        	if(this.containQueue(e.getQueue())){
				UrlConsumer consumer=null;
				for(int i=1;i<=config.getConsumerCount();i++){
					try {
						consumer = new UrlConsumer(e.getQueue());
					} catch (Exception e1) {
						logger.error(e1.getMessage(),e1);
						continue;
					} 
					context.getAutowireCapableBeanFactory().autowireBean(consumer);
					consumer.endPointName=e.getQueue();
					consumer.setConnection(this.connection);
					consumer.setUrl(e.getUrl());
					if(!"".equals(e.getMapName())){
						DataMapping map=(DataMapping)context.getBean(e.getMapName());
						consumer.setDataMapping(map);
					}
					Runnable listener=(Runnable)consumer;
					Thread consumerThread = new Thread(listener);
				    consumerThread.start();
				    logger.info("--start consumer:"+consumer.endPointName);
				}
			}
        }
	}
	private Connection getConnection() throws IOException, TimeoutException{
		if(this.connection==null){
			connection = newConnection();
		}
		return this.connection;
	}
	
	private Connection newConnection() throws IOException, TimeoutException{
		Connection n;
		logger.info("--mq hosts:"+config.getHost());
		String[] hosts=config.getHost().split(",");
//		String[] ports=config.getPort().split(",");
		Address[] addrs=new Address[hosts.length];
		for(int i=0;i<hosts.length;i++){
			Address addr=new Address(hosts[i],Integer.parseInt(config.getPort()));
			addrs[i]=addr;
		}
		n = getConnectionFactory().newConnection(addrs);
		return n;
	}

	
	@Override
	public void onApplicationEvent(ContextRefreshedEvent event) {
		if(!inited){
			logger.info("-------startListener-----");
	        startListener();
	        this.inited=true;
		}
	}
}
